package org.lql.chaos.kafka.example.partition;

import org.apache.kafka.clients.producer.internals.StickyPartitionCache;
import org.apache.kafka.common.Cluster;
import org.lql.chaos.kafka.example.enums.PartitionEnum;

/**
 * @author: liangqinglong
 * @date: 2025-08-10 13:11
 * @description: hash 分区策略
 **/
public class HashPartitionStrategy implements CustomPartitionStrategy {

	@Override
	public String getName() {
		return PartitionEnum.HASH.getName();
	}

	@Override
	public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {
		Integer count = cluster.partitionCountForTopic(topic);
		if (key == null) {
			// Utils.toPositive(Utils.murmur2(keyBytes)) % count;
			return new StickyPartitionCache().partition(topic, cluster); // 使用粘性分区缓存
		}
		return Math.abs(key.hashCode()) % count;
	}
}
